package com.zhao;

import org.apache.pulsar.client.api.*;

import java.util.concurrent.TimeUnit;

/**
 *
 * @return
 * @exception
 * @author guojiangtao
 * @date 2020/11/2 17:10
 */

public class ConsumerDemo {

    /**
     * Pulsar集群中broker的serviceurl
     */
    private static final String BROKER_SERVICEURL = "pulsar://182.61.36.149:6650";
    /**
     * 需要订阅的topic name
     */
    private static final String TOPIC_NAME = "persistent://public/default/my-topic20201102";
    // private static final String topicName = "persistent://public/default/my-topic";
    /**
     * 订阅名
     */
    private static final String SUBSCRIPTION_NAME = "my-sub";

    public static void main(String[] args) throws PulsarClientException {
        //构造Pulsar client
        PulsarClient client = PulsarClient.builder()
                .serviceUrl(BROKER_SERVICEURL)
                .build();
        //创建consumer
        Consumer<byte[]> consumer = client.newConsumer()
                .topic(TOPIC_NAME)
                .subscriptionName(SUBSCRIPTION_NAME)
                //指定消费模式，包含：Exclusive，Failover，Shared，Key_Shared。默认Exclusive模式
                .subscriptionType(SubscriptionType.Exclusive)
                //指定从哪里开始消费，还有Latest，valueof可选，默认Latest
                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
                //指定消费失败后延迟多久broker重新发送消息给consumer，默认60s
                .negativeAckRedeliveryDelay(60, TimeUnit.SECONDS)
                .subscribe();

        //消费消息
        while (true) {
            Message<byte[]> message = consumer.receive();
            try {
                System.out.printf("Message received: %s%n", new String(message.getData()));
                consumer.acknowledge(message);
            } catch (Exception e) {
                e.printStackTrace();
                consumer.negativeAcknowledge(message);
            }

        }
    }


}
